import 'dart:async';
import 'dart:math';

import 'package:flutter/material.dart';

void main() => runApp(MyApp());

class MyApp extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Stream',
      theme: ThemeData(
        primarySwatch: Colors.blue,
        splashColor: Colors.transparent,
      ),
      home: HomePage(),
    );
  }
}

class HomePage extends StatefulWidget {
  @override
  _HomePageState createState() => _HomePageState();
}

class _HomePageState extends State<HomePage> {
  StreamSubscription _streamSubscription;

  // 控制管理Stream
  StreamController<String> _streamController;

  // 使用sink往Controller添加数据
  StreamSink sink;

  // 以上使用的Stream只能订阅一次，创建可以多次订阅的Stream

  @override
  void initState() {
    super.initState();
    // _createStreamFromFuture();
    // _createStreamByController();
    _createStreamByBroadcast();
  }

  @override
  void dispose() {
    // 关闭Stream
    _streamController.close();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('Stream'),
      ),
      body: Center(
        child: Column(
          children: [
            Container(
              alignment: Alignment.center,
              padding: EdgeInsets.all(12),
              // 数据改变直接更新UI
              // 传统做法定义变量，当数据改变setState通知更新UI
              child: StreamBuilder<String>(
                stream: _streamController.stream,
                initialData: "...",
                builder: (ctx, snapshot) {
                  return Text(snapshot.data);
                },
              ),
            ),
            TextButton(
                onPressed: _addDataByController,
                child: Text('add data to Stream')),
            TextButton(
                onPressed: _addDataBySink, child: Text('add data by sink')),
            TextButton(onPressed: _pauseStream, child: Text('pause Stream')),
            TextButton(onPressed: _resumeStream, child: Text('resume Stream')),
            TextButton(onPressed: _cancelStream, child: Text('cancel Stream')),
          ],
        ),
      ),
    );
  }

  // 通过Future直接创建Stream
  void _createStreamFromFuture() {
    Stream<String> stream = Stream.fromFuture(fetchData());
    // 监听Stream
    _streamSubscription =
        stream.listen(onData, onError: onError, onDone: onDone);
    print('Stream Init completed.');
  }

  // 通过StreamController常见Stream
  void _createStreamByController() {
    _streamController = StreamController<String>();
    sink = _streamController.sink;
    _streamSubscription = _streamController.stream
        .listen(onData, onError: onError, onDone: onDone);

    // 多次订阅报错
    // _streamController.stream.listen(onData2);
  }

  // 创建可以被多次订阅的Stream
  void _createStreamByBroadcast() {
    _streamController = StreamController.broadcast();

    // 可以被多次订阅
    sink = _streamController.sink;
    _streamSubscription = _streamController.stream
        .listen(onData, onError: onError, onDone: onDone);

    _streamController.stream.listen(onData2);
  }

  // 通过StreamController直接添加数据
  void _addDataByController() async {
    String data = await fetchData();
    _streamController.add(data);
  }

  // 通过sink添加数据
  void _addDataBySink() async {
    String data = await fetchData();
    sink.add(data);
  }

  void onData(String data) {
    print('Data:$data');
  }

  void onData2(String data) {
    print('onData2:$data');
  }

  void onError(error) {
    print('Error:$error');
  }

  void onDone() {
    print('Done');
  }

  void _pauseStream() {
    _streamSubscription.pause();
  }

  void _resumeStream() {
    _streamSubscription.resume();
  }

  void _cancelStream() {
    _streamSubscription.cancel();
  }

  Future<String> fetchData() async {
    await Future.delayed(Duration(seconds: 1));
    // 模拟抛出异常
    // throw 'error throw';
    return "hello world:${Random().nextInt(255)}";
  }
}
